package com.siteseat.crudeq.client;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.ConcurrentModificationException;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;

import com.siteseat.common.foundation.Component;
import com.siteseat.common.foundation.PropertyHelper;
import com.siteseat.crudeq.client.v1.request.V1CommandConnection;
import com.siteseat.crudeq.client.v1.request.V1SubscriptionConnection;

/**
 *
 */
public class Client extends Component {

  private SubscriptionConnection subscriptionConnection;
  
  private CommandConnection commandConnection;
    
  private ClientProfile profile;
  
  private LinkedList<ClientMessage> buffer = new LinkedList<ClientMessage>();
  
  /**
   *
   *
   * @param profile
   */
  public Client( ClientProfile profile ) {
    this.profile = profile;
  }
  
  /**
   * @throws Exception 
   * 
   */
  public void start() throws Exception {
    init();
  }
  
  /**
   * @throws Exception 
   * 
   */
  public void shutdown() throws Exception {
    destroy();
  }

  /**
   * {@inheritDoc}
   */
  @Override
  protected void doDestroy() throws Exception {
    getProfile().save();
  }

  /**
   * {@inheritDoc}
   */
  @Override
  protected void doInit() throws Exception {
    ClientProfile p = getProfile();
    PropertyHelper.ensureInitialized( log, p, "profile" );

    this.commandConnection = new V1CommandConnection( p );
  }
  
  /**
   * 
   *
   * @return
   * @throws IOException 
   */
  protected SubscriptionConnection getSubscriptionConnection() throws IOException {
    if( this.subscriptionConnection == null ) {
      this.subscriptionConnection = new V1SubscriptionConnection( getProfile() );
    }
    return this.subscriptionConnection;
  }
  
  /**
   * 
   *
   * @return
   */
  protected CommandConnection getCommandConnection() {
    return this.commandConnection;
  }
  
  /**
   * Gets the profile
   *
   * @return the profile
   */
  public ClientProfile getProfile() {
    return profile;
  }

  /**
   * 
   */
  public void disconnect() {
    ensureInited();
    
    if( this.subscriptionConnection != null && this.subscriptionConnection.isConnected() ) {
      try {
        this.subscriptionConnection.disconnect();
      }catch( Exception e ) {
        this.log.warn( "Error during disconnect subscription", e );
      }
    }
    
    if( this.commandConnection.isConnected() ) {
      try {
        this.commandConnection.disconnect();
      }catch( Exception e ) {
        this.log.warn( "Error during disconnect command", e );
      }
    }
  }

  /**
   * 
   *
   * @param queueId
   * @param payload
   * @return
   * @throws IOException
   */
  public PublishResult click( int queueId, byte[] payload ) throws IOException {
    ensureInited();
    
    PublishResult r = getCommandConnection().click( queueId, payload );
    //TODO: record the send history
    return r;
  }

  /**
   * 
   *
   * @param queueId
   * @param payload
   * @return
   * @throws IOException
   */
  public PublishResult click( int queueId, String payload ) throws IOException {
    ensureInited();
    
    PublishResult r = getCommandConnection().click( queueId, payload );
    //TODO: record the send history
    return r;
  }

  /**
   * 
   *
   * @param mids
   * @return
   * @throws IOException
   */
  protected List<ClientMessage> historyWithSpecificOffset( Collection<QueueHistory> mids ) throws IOException {
    List<ClientMessage> t = getCommandConnection().history( mids );

    Map<Integer,QueueHistory> qh = getProfile().getReceiveHistory();
    for( ClientMessage m: t ) {
      QueueHistory mid = qh.get( m.getQueueId() );
      if( mid == null ) {
        throw new ConcurrentModificationException();
      }
      mid.setChunkId( m.getChunkId() );
      mid.setOffset( m.getNextPayloadBase() );
    }

    return t;
  }
  
  /**
   * 
   *
   * @param qids
   * @return
   * @throws IOException
   */
  public List<ClientMessage> history( Set<Integer> qids ) throws IOException {
    ensureInited();
    
    Map<Integer,QueueHistory>  h = getProfile().getReceiveHistory();
    for( Integer qid: qids ) {
      if( h.containsKey( qid ) == false ) {
        h.put( qid, new QueueHistory( qid ) );
      }
    }
    
    return historyWithSpecificOffset( h.values() );
  }
  
  /**
   * 
   *
   * @param qids
   * @return
   * @throws IOException
   */
  public List<ClientMessage> history( int... queueIdArray ) throws IOException {
    ensureInited();
    
    Set<Integer> qids = new HashSet<Integer>( queueIdArray.length );
    for( int qid: queueIdArray ) {
      qids.add( Integer.valueOf( qid ) );
    }
    return history( qids );
  }
  
  /**
   * 
   *
   * @return
   * @throws IOException
   */
  @SuppressWarnings( "unchecked" )
  public List<ClientMessage> history() throws IOException {
    ensureInited();
    
    return history( Collections.EMPTY_SET );
  }

  /**
   * 
   *
   * @param userName
   * @param password
   * @param clientId
   * @param language
   * @throws IOException
   */
  //public void login( String userName, String password, String clientId, String language ) throws IOException {
  //  getConnection().login( userName, password, clientId, language );
  //}

  /**
   * 
   *
   * @return
   */
  public boolean isConnected() {
    ensureInited();
    
    if( getCommandConnection().isConnected() ) {
      return false;
    }
    
    if( this.subscriptionConnection != null ) {
      return this.subscriptionConnection.isConnected();
    }
    
    return true;
  }

  /**
   * 
   *
   * @return
   */
  public boolean isLogined() {
    ensureInited();
    
    if( getCommandConnection().isLogined() ) {
      return false;
    }
    
    if( this.subscriptionConnection != null ) {
      return this.subscriptionConnection.isLogined();
    }
    
    return true;
  }

  /**
   * 
   *
   * @throws IOException
   */
  public void startWaitMessage() throws IOException {
    ensureInited();
    
    getSubscriptionConnection().startWaitMessage();
  }

  /**
   * 
   *
   * @throws IOException
   * @throws InterruptedException
   */
  public void stopWaitMessage() throws IOException, InterruptedException {
    ensureInited();
    
    getSubscriptionConnection().stopWaitMessage();
  }

  /**
   * 
   *
   * @param queueIds
   * @throws IOException
   */
  public void subscribe( Set<Integer> queueIds ) throws IOException {
    ensureInited();
    
    getSubscriptionConnection().subscribe( queueIds );
  }
  
  /**
   * 
   *
   * @param queueIds
   * @throws IOException
   */
  public void subscribe( Integer... queueIds ) throws IOException {
    Set<Integer> qidSet = new HashSet<Integer>( queueIds.length );
    for( Integer qid: queueIds ) {
      qidSet.add( qid );
    }
    
    subscribe( qidSet );
  }

  /**
   * 
   *
   * @param queueIds
   * @throws IOException
   */
  public void unsubscribe( Set<Integer> queueIds ) throws IOException {
    ensureInited();
    
    getSubscriptionConnection().unsubscribe( queueIds );
  }
  
  /**
   * 
   *
   * @param queueIds
   * @throws IOException
   */
  public void unsubscribe( Integer... queueIds ) throws IOException {
    Set<Integer> qidSet = new HashSet<Integer>( queueIds.length );
    for( Integer qid: queueIds ) {
      qidSet.add( qid );
    }
    
    unsubscribe( qidSet );
  }

  /**
   * 
   *
   * @throws IOException
   */
  public void unsubscribeAll() throws IOException {
    ensureInited();
    
    getSubscriptionConnection().unsubscribeAll();
  }

  /**
   * 
   *
   * @param timeoutBySeconds
   * @return
   * @throws InterruptedException
   * @throws IOException
   */
  public List<ClientMessage> waitNextMessage() throws InterruptedException, IOException {
    ensureInited();

    ClientMessage m;
    
    if( this.buffer.isEmpty() ) {
      m = getSubscriptionConnection().waitNextMessage();
      if( m == null ) {
        return null;
      }
      this.buffer.add( m );
    }

    m = this.buffer.getFirst();
    if( m == ClientMessage.STOPPED ) {
      this.buffer.removeFirst();
      return ClientMessage.STOPPED_LIST;
    }

    Map<Integer,QueueHistory> histories = getProfile().getReceiveHistory();
    QueueHistory h = histories.get( m.getQueueId() );
    if( h == null ) {
      throw new ConcurrentModificationException();
    }

    long expectedChunkId = h.getChunkId();

    if( m.isValid() == false ) {
      log.error( "Invalid message found. Jump to next chunk. Message: " + m );
      this.buffer.removeFirst();
      h.setChunkId( expectedChunkId + 1 );
      h.setOffset( 0 );
      return null;
    }

    long actualChunkId = m.getChunkId();
    int actualPayloadBase = m.getPayloadBase();     

    int expectedPayloadBase = h.getOffset();

    if( expectedChunkId == actualChunkId ) {
      // still in the same chunk
      if( expectedPayloadBase == actualPayloadBase ) {
        h.setOffset( m.getNextPayloadBase() );
        List<ClientMessage> r = new ArrayList<ClientMessage>(1);
        this.buffer.removeFirst();
        r.add( m );
        return r;
      }
      if( expectedPayloadBase < actualPayloadBase ) {
        // sync with history
        log.info( "Sync history to catch latest message: " + m );
        return historyWithSpecificOffset( histories.values() );
      } 
      
      log.info( "Ignore the duplicated message: " + m );
      this.buffer.removeFirst();
      return null;
    }
    else if( expectedChunkId < actualChunkId ) {
      // sync with history
      log.info( "Sync history to catch latest message: " + m );
      return historyWithSpecificOffset( histories.values() );
    }
    
    // ignore the duplicated message
    log.info( "Ignore the duplicated message: " + m );
    this.buffer.removeFirst();
    return null;
  }

}
